﻿using DotNetty.Codecs;
using DotNetty.Transport.Bootstrapping;
using DotNetty.Transport.Channels;
using DotNetty.Transport.Channels.Sockets;
using DotNetty.Buffers;
using Microsoft.Extensions.Logging;
using System;
using System.Net;
using System.Threading.Tasks;
namespace Athena.Transport.DotNetty
{
    using ChannelHandlerAdapers;
    using Codec;
    using Message;

    public class DotNettyTransportMessageListener : ITransportMessageListener, IDisposable
    {
        private readonly IRpcMessageDecoder _rpcMessageDecoder;
        private readonly IRpcMessageEncoder _rpcMessageEncoder;
        private readonly ILogger _logger;
        private IChannel _channel;

        public event ReceivedDelegate Received;

        public DotNettyTransportMessageListener(IRpcMessageCodecFactory rpcMessageCodecFactory, ILogger<DotNettyTransportMessageListener> logger)
        {
            _logger = logger;
            _rpcMessageDecoder = rpcMessageCodecFactory.CreateDecoder();
            _rpcMessageEncoder = rpcMessageCodecFactory.CreateEncoder();
        }

        public async Task StartAsync(EndPoint endPoint)
        {
            if (_logger.IsEnabled(LogLevel.Debug))
            {
                _logger.LogDebug($"开始启动服务器主机，监听地址：{endPoint}。");
            }
            var bossGroup = new MultithreadEventLoopGroup(1);
            var workerGroup = new MultithreadEventLoopGroup();
            var bootstrap = new ServerBootstrap();
            bootstrap.Group(bossGroup, workerGroup)
                .Channel<TcpServerSocketChannel>()
                .Option(ChannelOption.Allocator, PooledByteBufferAllocator.Default)
                .ChildHandler(new ActionChannelInitializer<ISocketChannel>(channel =>
                {
                    var pipeline = channel.Pipeline;
                    pipeline.AddLast(new LengthFieldPrepender(4));
                    pipeline.AddLast(new LengthFieldBasedFrameDecoder(int.MaxValue, 0, 4, 0, 4));
                    pipeline.AddLast(new TransportMessageChannelHandlerAdaper(_rpcMessageDecoder));
                    pipeline.AddLast(new ServerChannelHandlerAdaper(async (context, message) =>
                    {
                        var sender = new DotNettyTransportMessageServerSender(_rpcMessageEncoder, context);
                        await OnReceived(sender, message);
                    }, _logger));
                }));

            _channel = await bootstrap.BindAsync(endPoint);
            if (_logger.IsEnabled(LogLevel.Debug))
            {
                _logger.LogDebug($"服务主机启动成功，监听地址{endPoint}");
            }
        }

        public void Dispose()
        {
            Task.Run(async () =>
            {
                await _channel.DisconnectAsync();
            }).Wait();
        }

        public async Task OnReceived(ITransportMessageSender transportMessageSender, RpcTransportMessage rpcTransportMessage)
        {
            if (Received == null)
            {
                return;
            }
            await Received(transportMessageSender, rpcTransportMessage);
        }
    }
}
